package discovery

import (
	"context"
	"emailcenter/utils/logx"
	"emailcenter/utils/secret"
	"go.etcd.io/etcd/client/v3"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"sync"
	"time"
)

var xWRLock = sync.RWMutex{}

type Explorer struct {
	EtcdAddr    []string
	DialTimeout time.Duration

	client    *clientv3.Client
	closeCh   chan struct{}
	watchCh   clientv3.WatchChan
	keyPrefix string

	token         *secret.Authentication
	addrArray     []*grpc.ClientConn
	sourceWeight  []int
	currentWeight []int
	totalWeight   int
}

// NewExplorer 新建 Explorer 对象
func NewExplorer(etcdAddr []string, dialTimeout time.Duration) *Explorer {
	return &Explorer{
		EtcdAddr:    etcdAddr,
		DialTimeout: dialTimeout,
	}
}

// Stop 关闭发现
func (e *Explorer) Stop() {
	e.closeCh <- struct{}{}
}

// Start 开启发现
func (e *Explorer) Start(topic string, token *secret.Authentication) error {
	var err error

	e.client, err = clientv3.New(clientv3.Config{
		Endpoints:   e.EtcdAddr,
		DialTimeout: e.DialTimeout,
	})
	if err != nil {
		return err
	}

	e.keyPrefix = BuildPrefix(&Server{
		Name: topic,
	})
	e.closeCh = make(chan struct{})
	e.token = token

	if err = e.sync(); err != nil {
		return err
	}

	go e.watch()

	return nil
}

// sync 获取服务列表
func (e *Explorer) sync() error {
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	res, err := e.client.Get(ctx, e.keyPrefix, clientv3.WithPrefix())
	if err != nil {
		return err
	}

	e.clear()

	for _, v := range res.Kvs {
		info, err := ParseValue(v.Value)
		if err != nil {
			continue
		}
		e.append(info)
	}

	return nil
}

// watch 监听 kv 变化
func (e *Explorer) watch() {
	slowTicker := time.NewTicker(10 * time.Minute)
	quickTicker := time.NewTicker(time.Second)
	e.watchCh = e.client.Watcher.Watch(context.Background(), e.keyPrefix, clientv3.WithPrefix())

	for {
		if e.totalWeight == 0 {
			select {
			case <-e.closeCh:
				logx.LogrusObj.Infoln("[DISCOVERY] watch closed")
				return
			case res := <-e.watchCh:
				e.update(res.Events)
			case <-quickTicker.C:
				if err := e.sync(); err != nil {
					logx.LogrusObj.Error("[DISCOVERY] sync failed, error: ", err)
				}
			}
		} else {
			select {
			case <-e.closeCh:
				logx.LogrusObj.Infoln("[DISCOVERY] watch closed")
				return
			case res := <-e.watchCh:
				e.update(res.Events)
			case <-slowTicker.C:
				if err := e.sync(); err != nil {
					logx.LogrusObj.Error("[DISCOVERY] sync failed, error: ", err)
				}
			}
		}
	}
}

// update 更新服务列表
func (e *Explorer) update(events []*clientv3.Event) {
	for _, ev := range events {
		var info *Server
		var err error

		switch ev.Type {
		case clientv3.EventTypePut:
			info, err = ParseValue(ev.Kv.Value)
			if err != nil {
				continue
			}

			if !e.exist(info) {
				e.append(info)
			}
		case clientv3.EventTypeDelete:
			info, err = SplitPath(string(ev.Kv.Key))
			if err != nil {
				continue
			}
			_ = e.remove(info)
		default:
			logx.LogrusObj.Warning("[DISCOVERY] bad event: ", ev)
		}
	}
}

func (e *Explorer) append(server *Server) {
	conn, _ := grpc.Dial(server.Addr, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithPerRPCCredentials(e.token))

	e.totalWeight += server.Weight
	e.addrArray = append(e.addrArray, conn)
	e.sourceWeight = append(e.sourceWeight, server.Weight)
	e.currentWeight = append(e.currentWeight, server.Weight)
}

// exist 判断这个服务地址是否已经存在，防止服务访问冲突
func (e *Explorer) exist(server *Server) bool {
	for i := range e.addrArray {
		if e.addrArray[i].Target() == server.Addr {
			return true
		}
	}

	return false
}

// remove 移除服务
func (e *Explorer) remove(server *Server) bool {
	for i := range e.addrArray {
		if e.addrArray[i].Target() == server.Addr {
			e.addrArray[i].Close()
			e.totalWeight -= server.Weight
			e.addrArray[i] = e.addrArray[len(e.addrArray)-1]
			e.addrArray = e.addrArray[:len(e.addrArray)-1]
			e.sourceWeight[i] = e.sourceWeight[len(e.sourceWeight)-1]
			e.sourceWeight = e.sourceWeight[:len(e.sourceWeight)-1]
			e.currentWeight[i] = e.currentWeight[len(e.currentWeight)-1]
			e.currentWeight = e.currentWeight[:len(e.currentWeight)-1]
			return true
		}
	}

	return false
}

// clear 初始化、清空列表
func (e *Explorer) clear() bool {
	e.totalWeight = 0
	e.addrArray = []*grpc.ClientConn{}
	e.sourceWeight = []int{}
	e.currentWeight = []int{}

	return false
}

func (e *Explorer) Get() *grpc.ClientConn {

	var cMax = -99
	var cMaxIdx = -99

	xWRLock.RLock()

	if len(e.addrArray) == 0 {
		xWRLock.RUnlock()
		return nil
	}

	for idx, cw := range e.currentWeight {
		if cw >= cMax {
			cMax = cw
			cMaxIdx = idx
		}
		e.currentWeight[idx] += e.currentWeight[idx]
	}
	if cMaxIdx < 0 {
		xWRLock.RUnlock()
		return nil
	}
	e.currentWeight[cMaxIdx] -= e.totalWeight

	xWRLock.RUnlock()

	return e.addrArray[cMaxIdx]
}

// Sad 如果出现使用失败的情况，降权重
func (e *Explorer) Sad(addr string) {
	xWRLock.RLock()
	for idx := range e.addrArray {
		if e.addrArray[idx].Target() == addr {
			e.sourceWeight[idx]--
			e.totalWeight--
			break
		}
	}
	xWRLock.RUnlock()
}
